package cn.sheep.violet.tag

import cn.sheep.violet.config.ConfigHandler
import cn.sheep.violet.utils.{Jpools, TagUtils}
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
  * author: old sheep
  * QQ: 64341393 
  * Created 2018/10/19
  */
object TagContext {

    def main(args: Array[String]): Unit = {

        val sparkConf = new SparkConf()
            .setAppName(this.getClass.getSimpleName)
            .setMaster("local[*]")
        val sc = new SparkContext(sparkConf)
        val sqlContext = new SQLContext(sc)

        // 读取文件parquet
        val parquetData = sqlContext.read.parquet(ConfigHandler.parquetFilePath)
            // 过滤掉为空的用户
            .filter(TagUtils.userIfFilterCondition)

        parquetData.mapPartitions(iter => {
            val tagMap = new mutable.HashMap[String, Map[String, Int]]()

            val jedis = Jpools.getJedis

            iter.foreach(row => {
                var currentLineMap = Map[String, Int]()

                // 获取用户id
                val userId = TagUtils.batainUserOneId(row)

                // 提取关键字段
                val adSpaceType = row.getAs[Int]("adspacetype")
                val adSpaceTypeName = row.getAs[String]("adspacetypename")

                // 提取关键字段
                val appId = row.getAs[String]("appid")
                val appName = row.getAs[String]("appname")

                // 渠道
                val channelId = row.getAs[Int]("adplatformproviderid")
                val client = row.getAs[Int]("client")
                val network = row.getAs[String]("networkmannername")
                val ispName = row.getAs[String]("ispname")
                val keywords = row.getAs[String]("keywords")

                val pName = row.getAs[String]("provincename")
                val cName = row.getAs[String]("cityname")


                // 广告位类型
                if (adSpaceType > 9) currentLineMap += ("LC"+adSpaceType -> 1) else if (adSpaceType > 0) currentLineMap +=(("LC"+adSpaceType, 1))
                if (StringUtils.isNotEmpty(adSpaceTypeName)) currentLineMap += (("LN"+adSpaceTypeName, 1))

                if (StringUtils.isEmpty(appName)) {
                    if (StringUtils.isNotEmpty(appId)) {
                        val appId2Name = jedis.hget("appdict", appId)
                        if (StringUtils.isNotEmpty(appId2Name)) currentLineMap += ("APP"+appId2Name -> 1)
                    }
                } else currentLineMap += ("APP"+appName -> 1)


                if (channelId > 0) currentLineMap += ("CN"+channelId -> 1)
                client match {
                    case 1 => currentLineMap += (("D00010001", 1))
                    case 2 => currentLineMap += (("D00010002", 1))
                    case 3 => currentLineMap += (("D00010003", 1))
                    case _ =>
                }

                network.toUpperCase match  {
                    case "WIFI" => currentLineMap += (("D00020001", 1))
                    case "4G" => currentLineMap += (("D00020002", 1))
                    case "3G" => currentLineMap += (("D00020003", 1))
                    case "2G" => currentLineMap += (("D00020004", 1))
                    case _ =>
                }


                ispName.toUpperCase match  {
                    case "移动" => currentLineMap += (("D00030001", 1))
                    case "联通" => currentLineMap += (("D00030002", 1))
                    case "电信" => currentLineMap += (("D00030003", 1))
                    case _ =>
                }

                keywords.split("\\|")
                    .filter(kw => kw.size >= 3 && kw.size <= 8)
                    .foreach(word => currentLineMap += (("K"+word, 1)))

                if (StringUtils.isNotEmpty(pName)) currentLineMap += (("ZP"+pName, 1))
                if (StringUtils.isNotEmpty(cName)) currentLineMap += (("ZC"+cName, 1))

                tagMap.put(userId, currentLineMap)
            })

            jedis.close()
            tagMap.toIterator
        }).reduceByKey{
            (map1, map2) => (map1.toList ++ map2.toList).groupBy(_._1).mapValues(list => list.map(_._2).sum)
        }.foreach(println)

        sc.stop()

    }

}
